package ma.zhiqiang.dataline.core.convert;

import cn.hutool.json.JSONArray;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import lombok.AllArgsConstructor;
import lombok.NoArgsConstructor;
import ma.zhiqiang.dataline.core.dto.Column;
import ma.zhiqiang.dataline.core.dto.Schema;
import ma.zhiqiang.dataline.core.dto.SourceRecord;
import ma.zhiqiang.dataline.util.ClassTypeUtil;

import java.util.List;
import java.util.Properties;

public class RecordConvertJson implements RecordConvert{
    Properties properties;
    List<Schema> schemas;

    public RecordConvertJson(Properties properties) {
        this.properties = properties;
        this.schemas = JSONUtil.toList((JSONArray) properties.get("schema"), Schema.class);
    }

    @Override
    public SourceRecord convert(Object record) {
        String recordString = (String) record;
        JSONObject object = JSONUtil.parseObj(recordString);
        if(null==object){return null;}
        if(object.size() < schemas.size()){ throw new RuntimeException("schema集合长度应小于等于数据字段集合长度");}
        SourceRecord sourceRecord = new SourceRecord(schemas.size());
        schemas.stream().forEach(schema -> {
            String alias = schema.getAlias();
            Class type = ClassTypeUtil.getType(schema.getType());
            Object value = object.get(alias, type);
            sourceRecord.addColumn(new Column(alias,type,value),schema.getHandler());
        });
        return sourceRecord;
    }
}
